package stateful;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Value state 使用
 * 调用底层process方法
 *
 */
public class ValueStateDemo2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 开启checkpoint
        env.enableCheckpointing(5000);

        DataStreamSource<String> lines = env.socketTextStream("hadoop1", 8888);

        lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                for (String w : value.split(" ")) {
                    if (w.contains("error")) {
                        throw new RuntimeException("数据出错!");
                    } else {
                        out.collect(Tuple2.of(w, 1));
                    }
                }
            }
        }).keyBy(t -> t.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String,Integer>>() {

                    // 定义状态
                    private transient ValueState<Integer> values;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 初始化时恢复状态
                        // 1.定义一个状态描述器
                        ValueStateDescriptor<Integer> valueStateDescriptor = new ValueStateDescriptor<>("state", Integer.class);
                        // 2.从运行时上下文中获取状态数据
                        values = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                        // 获取状态
                        Integer historyValue = values.value();
                        Integer currentValue = value.f1;
                        Integer totalValue;
                        if (historyValue == null) {
                            historyValue = 0;
                        }
                        totalValue = historyValue + currentValue;
                        // 更新状态
                        values.update(totalValue);
                        out.collect(Tuple2.of(value.f0, totalValue));
                    }
                })
                .print();

        env.execute("");
    }
}
